import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.regionserver.BloomType;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
//https://cloud.tencent.com/developer/article/1336654   
public class HbaseClientDemo {

        Connection conn = null;
        Table table = null;
        @Before
        public void getConn() throws IOException {
            //构建一个连接对象
            Configuration conf = HBaseConfiguration.create();  //会自动加载hbase-site.xml
            //下面的地址不能用数字，只能用字母，否则不识别，
            //做法：在本地的C:\Windows\System32\drivers\etc的hosts下面写上映射
            // 192.168.88.20 hbase
            //192.168.88.21 hbase1
            //192.168.88.22 hbase2
            //如果没有hosts这个文件，则自己建一个
            conf.set("hbase.zookeeper.quorum", "hbase:2181,hbase1:2181,hbase2:2181");
            conn = ConnectionFactory.createConnection(conf);
            table = conn.getTable(TableName.valueOf("t_user1"));
        }
    /**
     * SQL语言共分为四大类：数据查询语言DQL，数据操纵语言DML，数据定义语言DDL，数据控制语言DCL
     * 1) DQL基本结构是由SELECT子句，FROM子句，WHERE子句组成的查询块
     * 2) DML主要有三种形式：插入：INSERT  更新：UPDATE  删除：DELETE
     * 3) DDL用来创建数据库中的各种对象-----表、视图、索引、同义词、聚簇等
     * 4) 数据控制语言DCL用来授予或回收访问数据库的某种特权，并控制数据库操纵事务发生的时间及效果，对数据库实行监视等。
     * <p>
     * <p>
     * DDL
     *
     * @throws IOException
     */
    /**
     * 3.6.1.DDL操作
     * 1、创建一个连接
     * Connection conn = ConnectionFactory.createConnection(conf);
     * 2、拿到一个DDL操作器：表管理器admin
     * Admin admin = conn.getAdmin();
     * 3、用表管理器的api去建表、删表、修改表定义
     * admin.createTable(HTableDescriptor descriptor);
     * @throws IOException
     */
        @Test
        public void teatCreateTable() throws IOException {
            //从连接中构造一个DDL操作器
            Admin admin = conn.getAdmin();
            //创建一个表定义描述对象
            HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf("t_user1"));
            //创建列族定义描述对象
            HColumnDescriptor hColumnDescriptor_1 = new HColumnDescriptor("base_info");
            HColumnDescriptor hColumnDescriptor_2 = new HColumnDescriptor("extra_info");
            //设置该列族中存储数据的最大版本数，默认是1
            hColumnDescriptor_1.setMaxVersions(3);

            //将列族定义信息对象放入表定义对象中
            hTableDescriptor.addFamily(hColumnDescriptor_1);
            hTableDescriptor.addFamily(hColumnDescriptor_2);

                //用DDL操作器对象：admin来建表
            admin.createTable(hTableDescriptor);

            //关闭连接
            admin.close();
            conn.close();

        }

    /**
     * 删除表
     */

    @Test
    public void testDropTable() throws IOException {
        Admin admin = conn.getAdmin();
        //停用表  意思是不能在对这个表进行操作
        admin.disableTable(TableName.valueOf("hbase"));
        //删除表
        admin.deleteTable(TableName.valueOf("t_user1"));

        admin.close();
        conn.close();
    }

    /**
     * 修改表定义--添加一个列族
     */
//    @Test
//    public void testAlterTable() throws IOException {
//        Admin admin = conn.getAdmin();
//
//        //取出旧的表定义信息
//        HTableDescriptor tableDescriptor = admin.getTableDescriptor(TableName.valueOf("t_user1"));
//
//        //新构造一个列族定义
//        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor("other_info");
//        //设置该列族的布隆过滤器类型
//        hColumnDescriptor.setBloomFilterType(BloomType.ROWCOL);
//        //将列族定义添加到表定义对象中
//        tableDescriptor.addFamily(hColumnDescriptor);
//        //将修改过的表定义交给admin去提交
//        admin.modifyTable(TableName.valueOf("t_user1"), tableDescriptor);
//
//        admin.close();
//        conn.close();
//    }
    /**
     * DML  数据的增删改查
     * <p>
     * 添加一条数据
     */

    @Test
    public void insertData() throws IOException {
        /**
         * Bytes.toBytes()为hadoop方法,String.getBytes()为java方法。
         *toBytes()方法是将参数使用UTF-8的编码格式转换成byte[],getBytes()是用读取file.encoding的编码格式,
         * 然后用读取的格式进行转换,所以,getBytes转换的byte[]的格式取决于操作系统和用户设置,不推荐这种方法,
         * 最好统一只用toBytes()方法
         */
        Put put = new Put(Bytes.toBytes("zhangsan"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("name"), Bytes.toBytes("lisi"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("age"), Bytes.toBytes("22"));
        table.put(put);
        System.out.println("添加成功");
    }
    /**
     * 批添加数据
     *
     * @throws IOException
     */
    @Test
    public void insertBatch() throws IOException {
        ArrayList<Put> arrayList = new ArrayList<Put>();
        for (int i = 45; i < 50; i++) {
            /**
             * 我们知道HBase中的Value默认是以byte数组存储的，我们通常使用String作为输入，
             * 那么需要调用Bytes.toBytes( )把String转换为byte数组，byte数组的每个元素为字符。
             * 打印需要使用new String( )把byte数组再次转换为String，若直接打印会是字符的ACSII码。
             *但是如果直接对byte数组的每个元素进行赋值 b[i]=0 ，每个元素是以\x00的形式存储的，但直接打印可以得到0。
             */
            Put put = new Put(Bytes.toBytes("1234" + i));
            put.addColumn(Bytes.toBytes("extra_info"), Bytes.toBytes("name"), Bytes.toBytes("wangwu" + i));
            put.addColumn(Bytes.toBytes("extra_info"), Bytes.toBytes("password"), Bytes.toBytes("1234" + i));
            arrayList.add(put);
        }

        table.put(arrayList);

    }
    /**
     * 修改数据 相同 rowkey 列簇 column 会覆盖新值
     */
    @Test
    public void updateData() throws IOException {
        Put put = new Put(Bytes.toBytes("zhangsan"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("name"), Bytes.toBytes("zhangsan2"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("age"), Bytes.toBytes("23"));
        table.put(put);

    }

    /**
     * 删除数据
     *
     * @throws IOException
     */
    @Test
    public void deleteData() throws IOException {
        //按行键删除
//        Delete delete = new Delete(Bytes.toBytes("zhangsan"));
//        table.delete(delete);

        //按列族删除  ：行键 列族
//        Delete delete1 = new Delete(Bytes.toBytes("123445"));
//        delete1.addFamily(Bytes.toBytes("extra_info"));
//        table.delete(delete1);

        //按cell删除 ：行键 列族 key
        Delete delete2 = new Delete(Bytes.toBytes("123447"));
        delete2.addColumns(Bytes.toBytes("extra_info"), Bytes.toBytes("name"));
        table.delete(delete2);
        System.out.println("删除数据成功");
    }
    /**
     * 查询数据
     *行键
     * @throws IOException
     */
    @Test
    public void queryData() throws IOException {
        Get get = new Get(Bytes.toBytes("123449"));
        Result result = table.get(get);
        System.out.println(Bytes.toInt(result.getValue(Bytes.toBytes("extra_info"), Bytes.toBytes("password"))));
        System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("extra_info"), Bytes.toBytes("name"))));
    }
    @Test
    public void scanData() throws IOException {
        Scan scan = new Scan();
         scan.addFamily(Bytes.toBytes("extra_info")); //扫描指定family

        // 扫描cell指定
        // scan.addColumn(Bytes.toBytes("extra_info"),
        // Bytes.toBytes("password"));

        // 按rowkey字典序扫描
        // scan.setStartRow(Bytes.toBytes("wangsf_0"));
        // scan.setStopRow(Bytes.toBytes("wangwu"))
        /**
         * 扫描操作不会通过一次RPC请求返回所有匹配的行，而是以行为单位进行返回。很明显，行的数目很大，可能有上千条甚至更多，
         * 同时在一次请求中发送大量数据，会占用大量的系统资源并消耗很长时间。
         * ResultScanner类把扫描操作转换为类似的get操作，它将每一行数据封装成一个Result实例，
         * 并将所有的Result实例放入一个迭代器中。ResultScanner的一些方法如下：
         */
        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            System.out.println(Bytes.toInt(result.getValue(Bytes.toBytes("extra_info"), Bytes.toBytes("password"))));
            System.out.println(Bytes.toString(result.getValue(Bytes.toBytes("extra_info"), Bytes.toBytes("name"))));
        }

    }
}

